package com.example.klinemqflink.fun;

import com.example.klinemqflink.model.Kline;
import com.example.klinemqflink.model.MarketData;
import com.example.klinemqflink.utils.TimeRoundingUtil;
import lombok.Data;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Data
public class KLineAggregateFunction implements AggregateFunction<MarketData, Kline, Kline> {
    private static final Logger logger = LoggerFactory.getLogger(KLineAggregateFunction.class);

    private String frequency;

    public KLineAggregateFunction(String frequency) {
        this.frequency = frequency;
    }

    @Override
    public Kline createAccumulator() {
        Kline kline = new Kline();
        kline.setFrequency(this.frequency);
        return kline;
    }

    @Override
    public Kline add(MarketData value, Kline kline) {
        if (kline.getStartTime() == 0) {
            kline.setChannel(value.getChannel());
            kline.setInstrument(value.getInstrument());
            kline.setSymbol(value.getSymbol());
            //处理整分钟
            long timestampRounding = TimeRoundingUtil.getRoundedTimestamp(value.getTimestamp(),this.frequency);
//            String s = LocalTimeUtil.timestampToDateTime(timestampRounding);
            kline.setStartTime(timestampRounding);
            kline.setOpen(value.getPrice());
            kline.setHigh(value.getPrice());
            kline.setLow(value.getPrice());
        }
        kline.setEndTime(value.getTimestamp());
        kline.setHigh(Math.max(kline.getHigh(), value.getPrice()));
        kline.setLow(Math.min(kline.getLow(), value.getPrice()));
        kline.setClose(value.getPrice());
        //过程bar
        return kline;
    }

    @Override
    public Kline getResult(Kline kline) {
        return kline;
    }

    @Override
    public Kline merge(Kline first, Kline second) {
        first.setHigh(Math.max(first.getHigh(), second.getHigh()));
        first.setLow(Math.min(first.getLow(), second.getLow()));
        first.setEndTime(second.getEndTime());
        first.setClose(second.getClose());
        return first;
    }
}
